package io.vertx.kue

import io.vertx.kue.queue.*
import io.vertx.kue.queue.JobState
import io.vertx.kue.queue.KueWorker
import io.vertx.kue.service.JobService
import io.vertx.kue.util.RedisHelper

import io.vertx.core.DeploymentOptions
import io.vertx.core.Future
import io.vertx.core.Handler
import io.vertx.core.Vertx
import io.vertx.core.eventbus.Message
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import io.vertx.redis.RedisClient
import io.vertx.redis.op.RangeLimitOptions
import java.util.Optional

import io.vertx.kue.queue.KueVerticle

/**
 * The Kue class refers to a job queue.
 *
 * @author Eric Zhao
 */
class Kue(private val vertx: Vertx, private val config: JsonObject) {
    /**
     * Get the JobService.
     * *Notice: only available in package scope*
     */
    internal val jobService: JobService
    private val client: RedisClient

    /**
     * Get the job types present.
     *
     * @return async result list
     */
    val allTypes: Future<List<String>>
        get() {
            val future = Future.future<List<String>>()
            jobService.getAllTypes(future.completer())
            return future
        }

    /**
     * Get queue work time in milliseconds.
     *
     * @return async result
     */
    val workTime: Future<Long>
        get() {
            val future = Future.future<Long>()
            jobService.getWorkTime(future.completer())
            return future
        }

    init {
        this.jobService = JobService.createProxy(vertx, KueVerticle.EB_JOB_SERVICE_ADDRESS)
        this.client = RedisHelper.client(vertx, config)
        Job.setVertx(vertx, RedisHelper.client(vertx, config)) // init static vertx instance inner job
    }

    /**
     * Create a job instance.
     *
     * @param type job type
     * @param data job extra data
     * @return a new job instance
     */
    fun createJob(type: String, data: JsonObject): Job {
        return Job(type, data)
    }

    private fun processInternal(type: String, handler: Handler<Job>, isWorker: Boolean) {
        val worker = KueWorker(type, handler, this)
        vertx.deployVerticle(worker, DeploymentOptions().setWorker(isWorker)) { r0 ->
            if (r0.succeeded()) {
                this.on<Any>("job_complete", Handler{ msg ->
                    val dur = Job((msg.body() as JsonObject).getJsonObject("job")).getDuration()
                    client.incrby(RedisHelper.getKey("stats:work-time"), dur) { r1 ->
                        if (r1.failed())
                            r1.cause().printStackTrace()
                    }
                })
            }
        }
    }

    /**
     * Queue-level events listener.
     *
     * @param eventType event type
     * @param handler   handler
     */
    fun <R> on(eventType: String, handler: Handler<Message<R>>): Kue {
        vertx.eventBus().consumer(Kue.workerAddress(eventType), handler)
        return this
    }

    /**
     * Process a job in asynchronous way.
     *
     * @param type    job type
     * @param n       job process times
     * @param handler job process handler
     */
    fun process(type: String, n: Int, handler: Handler<Job>): Kue {
        var n = n
        if (n <= 0) {
            throw IllegalStateException("The process times must be positive")
        }
        while (n-- > 0) {
            processInternal(type, handler, false)
        }
        setupTimers()
        return this
    }

    /**
     * Process a job in asynchronous way (once).
     *
     * @param type    job type
     * @param handler job process handler
     */
    fun process(type: String, handler: Handler<Job>): Kue {
        processInternal(type, handler, false)
        setupTimers()
        return this
    }

    /**
     * Process a job that may be blocking.
     *
     * @param type    job type
     * @param n       job process times
     * @param handler job process handler
     */
    fun processBlocking(type: String, n: Int, handler: Handler<Job>): Kue {
        var n = n
        if (n <= 0) {
            throw IllegalStateException("The process times must be positive")
        }
        while (n-- > 0) {
            processInternal(type, handler, true)
        }
        setupTimers()
        return this
    }

    // job logic

    /**
     * Get job from backend by id.
     *
     * @param id job id
     * @return async result
     */
    fun getJob(id: Long): Future<Optional<Job>> {
        val future = Future.future<Optional<Job>>()
        jobService.getJob(id, Handler{ r ->
            if (r.succeeded()) {
                future.complete(Optional.ofNullable(r.result()))
            } else {
                future.fail(r.cause())
            }
        })
        return future
    }

    /**
     * Remove a job by id.
     *
     * @param id job id
     * @return async result
     */
    fun removeJob(id: Long): Future<Void> {
        return this.getJob(id).compose { r ->
            if (r.isPresent()) {
                r.get().remove()
            } else {
                Future.succeededFuture()
            }
        }
    }

    /**
     * Judge whether a job with certain id exists.
     *
     * @param id job id
     * @return async result
     */
    fun existsJob(id: Long): Future<Boolean> {
        val future = Future.future<Boolean>()
        jobService.existsJob(id, future.completer())
        return future
    }

    /**
     * Get job log by id.
     *
     * @param id job id
     * @return async result
     */
    fun getJobLog(id: Long): Future<JsonArray> {
        val future = Future.future<JsonArray>()
        jobService.getJobLog(id, future.completer())
        return future
    }

    /**
     * Get a list of job in certain state in range (from, to) with order.
     *
     * @return async result
     * @see JobService.jobRangeByState
     */
    fun jobRangeByState(state: String, from: Long, to: Long, order: String): Future<List<Job>> {
        val future = Future.future<List<Job>>()
        jobService.jobRangeByState(state, from, to, order, future.completer())
        return future
    }

    /**
     * Get a list of job in certain state and type in range (from, to) with order.
     *
     * @return async result
     * @see JobService.jobRangeByType
     */
    fun jobRangeByType(type: String, state: String, from: Long, to: Long, order: String): Future<List<Job>> {
        val future = Future.future<List<Job>>()
        jobService.jobRangeByType(type, state, from, to, order, future.completer())
        return future
    }

    /**
     * Get a list of job in range (from, to) with order.
     *
     * @return async result
     * @see JobService.jobRange
     */
    fun jobRange(from: Long, to: Long, order: String): Future<List<Job>> {
        val future = Future.future<List<Job>>()
        jobService.jobRange(from, to, order, future.completer())
        return future
    }

    // runtime cardinality metrics

    /**
     * Get cardinality by job type and state.
     *
     * @param type  job type
     * @param state job state
     * @return corresponding cardinality (Future)
     */
    fun cardByType(type: String, state: JobState): Future<Long> {
        val future = Future.future<Long>()
        jobService.cardByType(type, state, future.completer())
        return future
    }

    /**
     * Get cardinality by job state.
     *
     * @param state job state
     * @return corresponding cardinality (Future)
     */
    fun card(state: JobState): Future<Long> {
        val future = Future.future<Long>()
        jobService.card(state, future.completer())
        return future
    }

    /**
     * Get cardinality of completed jobs.
     *
     * @param type job type; if null, then return global metrics.
     */
    fun completeCount(type: String): Future<Long> {
        val future = Future.future<Long>()
        jobService.completeCount(type, future.completer())
        return future
    }

    /**
     * Get cardinality of failed jobs.
     *
     * @param type job type; if null, then return global metrics.
     */
    fun failedCount(type: String): Future<Long> {
        val future = Future.future<Long>()
        jobService.failedCount(type, future.completer())
        return future
    }

    /**
     * Get cardinality of inactive jobs.
     *
     * @param type job type; if null, then return global metrics.
     */
    fun inactiveCount(type: String): Future<Long> {
        val future = Future.future<Long>()
        jobService.inactiveCount(type, future.completer())
        return future
    }

    /**
     * Get cardinality of active jobs.
     *
     * @param type job type; if null, then return global metrics.
     */
    fun activeCount(type: String): Future<Long> {
        val future = Future.future<Long>()
        jobService.activeCount(type, future.completer())
        return future
    }

    /**
     * Get cardinality of delayed jobs.
     *
     * @param type job type; if null, then return global metrics.
     */
    fun delayedCount(type: String): Future<Long> {
        val future = Future.future<Long>()
        jobService.delayedCount(type, future.completer())
        return future
    }

    /**
     * Return job ids with the given `state`.
     *
     * @param state job state
     * @return async result list
     */
    fun getIdsByState(state: JobState): Future<List<Long>> {
        val future = Future.future<List<Long>>()
        jobService.getIdsByState(state, future.completer())
        return future
    }

    /**
     * Set up timers for checking job promotion and active job ttl.
     */
    private fun setupTimers() {
        this.checkJobPromotion()
        this.checkActiveJobTtl()
    }

    /**
     * Check job promotion.
     * Promote delayed jobs, checking every `ms`.
     */
    private fun checkJobPromotion() { // TODO: TO REVIEW
        val timeout = config.getInteger("job.promotion.interval", 1000)!!
        val limit = config.getInteger("job.promotion.limit", 1000)!!
        // need a mechanism to stop the circuit timer
        vertx.setPeriodic(timeout.toLong()) { l ->
            client.zrangebyscore(RedisHelper.getKey("jobs:DELAYED"), 0.toString(), System.currentTimeMillis().toString(),
                    RangeLimitOptions(JsonObject().put("offset", 0).put("count", limit))) { r ->
                if (r.succeeded()) {
                    r.result().forEach { r1 ->
                        val id = java.lang.Long.parseLong(RedisHelper.stripFIFO(r1 as String))
                        this.getJob(id).compose { jr -> jr.get().inactive() }
                                .setHandler({ jr ->
                                    if (jr.succeeded()) {
                                        jr.result().emit("promotion", jr.result().getId())
                                    } else {
                                        jr.cause().printStackTrace()
                                    }
                                })
                    }
                } else {
                    r.cause().printStackTrace()
                }
            }
        }
    }

    /**
     * Check active job ttl.
     */
    private fun checkActiveJobTtl() {  // TODO
        val timeout = config.getInteger("job.ttl.interval", 1000)!!
        val limit = config.getInteger("job.ttl.limit", 1000)!!
        // need a mechanism to stop the circuit timer
        vertx.setPeriodic(timeout.toLong()) { l ->
            client.zrangebyscore(RedisHelper.getKey("jobs:ACTIVE"), 100000.toString(), System.currentTimeMillis().toString(),
                    RangeLimitOptions(JsonObject().put("offset", 0).put("count", limit))) { r ->

            }
        }
    }

    companion object {

        /**
         * Generate handler address with certain job on event bus.
         *
         * Format: vertx.kue.handler.job.{handlerType}.{addressId}.{jobType}
         *
         * @return corresponding address
         */
        fun getCertainJobAddress(handlerType: String, job: Job): String {
            return "vertx.kue.handler.job." + handlerType + "." + job.getAddress_id() + "." + job.getType()
        }

        /**
         * Generate worker address on event bus.
         *
         * Format: vertx.kue.handler.workers.{eventType}
         *
         * @return corresponding address
         */
        fun workerAddress(eventType: String): String {
            return "vertx.kue.handler.workers." + eventType
        }

        /**
         * Generate worker address on event bus.
         *
         * Format: vertx.kue.handler.workers.{eventType}.{addressId}
         *
         * @return corresponding address
         */
        fun workerAddress(eventType: String, job: Job): String {
            return "vertx.kue.handler.workers." + eventType + "." + job.getAddress_id()
        }

        /**
         * Create a Kue instance.
         *
         * @param vertx  vertx instance
         * @param config config json object
         * @return kue instance
         */
        fun createQueue(vertx: Vertx, config: JsonObject): Kue {
            return Kue(vertx, config)
        }
    }

}
